package cn.doitedu.demo2;

import cn.doitedu.beans.EventBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Date;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/14
 * @Desc: 学大数据，上多易教育
 * 一个用户连续 7天发生A行为，则触发消息
 **/
public class RuleCalculator_03 implements RuleCalculator {

    ValueState<Integer> dayCntState;
    ValueState<String> lastDayState;
    String rule_id = "rule-03";
    JSONObject message;




    @Override
    public void init(RuntimeContext runtimeContext) {

        dayCntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(rule_id+"-days", Integer.class));
        lastDayState = runtimeContext.getState(new ValueStateDescriptor<String>(rule_id+"-last", String.class));


        message = new JSONObject();
        message.put("rule_id",rule_id);

    }

    @Override
    public void calc(EventBean eventBean, Collector<String> out) throws IOException {

        if (eventBean.getEvent_id().equals("A")) {

            String lastDay = lastDayState.value();
            long time = new Date().getTime();
            String today = DateFormatUtils.format(time, "yyyy-MM-dd");
            String yesterday = DateFormatUtils.format(time - 24 * 60 * 60 * 1000, "yyyy-MM-dd");


            if (lastDay != null && lastDay.equals(yesterday)) {
                dayCntState.update(dayCntState.value() == null ? 1 : dayCntState.value() + 1);

                // 判断是否已经到了7天
                if (dayCntState.value() == 7) {
                    message.put("user_id",eventBean.getUser_id());
                    message.put("hit_time",eventBean.getAction_time());
                    out.collect(message.toJSONString());
                }


            } else {
                if (!today.equals(lastDay)) {
                    dayCntState.update(1);
                }
            }

            lastDayState.update(today);

        }
    }

    @Override
    public String getRuleId() {
        return this.rule_id;
    }
}
